package com.partsoft.dits.mq;

import org.apache.commons.pool.KeyedObjectPool;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;

import com.ibm.mq.MQC;
import com.ibm.mq.MQException;
import com.ibm.mq.MQGetMessageOptions;
import com.ibm.mq.MQMessage;
import com.ibm.mq.MQQueue;
import com.ibm.mq.MQQueueManager;
import com.partsoft.dits.DataEdgeDescriptor;
import com.partsoft.dits.EdgeException;
import com.partsoft.dits.MonitorStatus;
import com.partsoft.dits.log.Log;
import com.partsoft.dits.mq.ContextConfigurableMQQueueMonitorReactor.MQQueueListenerContextImpl;
import com.partsoft.dits.utils.EdgeLogUtils;
import com.partsoft.dits.utils.ExceptionUtils;

public class MQQueueListenerWorker implements Runnable, InitializingBean {

	private MonitorStatus monitorStatus = MonitorStatus.STOPPED;

	private final MQQueueListenerContextImpl listenerContext;

	private MQGetMessageOptions mqGetMessageOptions = new MQGetMessageOptions();

	private int accessQueueFlags = MQC.MQOO_INPUT_SHARED | MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING
			| MQC.MQOO_INQUIRE;

	private void initMQGetMessageOptions() {
		mqGetMessageOptions.options = mqGetMessageOptions.options + MQC.MQGMO_SYNCPOINT;
		mqGetMessageOptions.options = mqGetMessageOptions.options + MQC.MQGMO_WAIT;
		mqGetMessageOptions.options = mqGetMessageOptions.options + MQC.MQGMO_FAIL_IF_QUIESCING;
		mqGetMessageOptions.waitInterval = getRetrieveInterval() * 1000;
	}

	public MQQueueListenerWorker(MQQueueListenerContextImpl ctx) {
		this.listenerContext = ctx;
		initMQGetMessageOptions();
	}

	public String getListenerQMName() {
		return getListenerContext().getQueueManager();
	}

	public String getListenerQueueName() {
		return getListenerContext().getQueue();
	}

	public int getRetrieveInterval() {
		return new Long(getListenerContext().getRetrieveInterval()).intValue();
	}

	public MonitorStatus getMonitorStatus() {
		synchronized (this) {
			return monitorStatus;
		}
	}

	public void setMonitorStatus(MonitorStatus monitorStatus) {
		synchronized (this) {
			this.monitorStatus = monitorStatus;
		}
	}

	public void run() {
		setMonitorStatus(MonitorStatus.STARTED);
		boolean errorFocus = false;
		while (getMonitorStatus() != MonitorStatus.STOPPED && Thread.currentThread().isInterrupted() == false) {
			if (getMonitorStatus() == MonitorStatus.PAUSED) {
				try {
					Thread.sleep(1000);
				} catch (Throwable e) {
				}
				continue;
			}
			try {
				Thread.sleep(1000);
			} catch (Throwable e) {
			}
			if (getMonitorStatus() == MonitorStatus.STOPPED || getMonitorStatus() == MonitorStatus.PAUSED
					|| Thread.currentThread().isInterrupted())
				continue;
			try {
				doMonitor();
				if (errorFocus) {
					EdgeLogUtils.warn(getListenerContext().getEdgeDescriptor(),
							String.format("监控队列(%s:%s)恢复正常", getListenerQMName(), getListenerQueueName()));
				}
				errorFocus = false;
			} catch (Throwable e) {
				if ( e instanceof MQException || ExceptionUtils.findCause(e, MQException.class) != null) {
					if (!errorFocus) {
						errorFocus = true;
						EdgeLogUtils.error(getListenerContext().getEdgeDescriptor(), e);
					}
				} else {
					EdgeLogUtils.error(getListenerContext().getEdgeDescriptor(), e);
				}
			}
		}
		setMonitorStatus(MonitorStatus.STOPPED);
	}

	public KeyedObjectPool getQmPool() {
		return getListenerContext().getQueueManagerPool();
	}

	public MQGetMessageOptions getMQGetMessageOptions() {
		return mqGetMessageOptions;
	}

	public int getAccessQueueFlags() {
		return accessQueueFlags;
	}

	protected void doMonitor() throws Exception {
		MQGetMessageOptions gmo = getMQGetMessageOptions();
		int acqFlags = getAccessQueueFlags();
		MQQueue queue = null;
		MQQueueManager qm = null;
		byte[] msgBuffer = null;
		String qName = getListenerQueueName();
		String qmName = getListenerQMName();
		boolean commitSupport = true;
		try {
			qm = (MQQueueManager) getQmPool().borrowObject(qmName);
			queue = qm.accessQueue(qName, acqFlags);
			while (queue.getCurrentDepth() > 0) {
				if (commitSupport)
					try {
						qm.begin();
					} catch (MQException e) {
						if (e.completionCode == 2 && e.reasonCode == 2012) {
							commitSupport = false;
						} else
							throw e;
					}
				MQMessage tMsg = new MQMessage();
				queue.get(tMsg, gmo);
				msgBuffer = new byte[tMsg.getMessageLength()];
				tMsg.readFully(msgBuffer);
				processXmlMessage(msgBuffer);
				if (commitSupport)
					qm.commit();
			}
		} catch (MQException e) {
			if (e.completionCode == 2 && e.reasonCode == 2033) {
				Log.debug(String.format("队列(%s:%s)无消息退出。", qmName, qName));
			} else {
				if (qm != null && commitSupport) {
					qm.backout();
				}
				throw e;
			}
		} finally {
			if (queue != null)
				try {
					queue.close();
				} catch (MQException e2) {
				}
			if (qm != null) {
				try {
					qm.disconnect();
				} catch (MQException e2) {
				}
				try {
					qm.close();
				} catch (Throwable e2) {
				}
				getQmPool().returnObject(qmName, qm);
			}
		}
	}

	private void processXmlMessage(byte[] sourceXML) {
		Log.info(String.format("线程(%s)收到MQ一条消息。", Thread.currentThread().getName()));
		boolean processed = false;
		String mqid = null;
		for (DataEdgeDescriptor descriptor : getListenerContext().getDataDescriptors()) {
			if (mqid == null && ClassUtils.isAssignableValue(MQQueueMonitorDescriptor.class, descriptor)) {
				mqid = ((MQQueueMonitorDescriptor) descriptor).getConnectPool();
			}
			try {
				listenerContext.receiveData(descriptor, sourceXML);
				processed = true;
			} catch (Throwable e) {
				Log.warn(new EdgeException(descriptor, e.getMessage()));
			}
		}

		if (!processed) {
			MQPoolDescriptor mqp = new MQPoolDescriptor();
			mqp.setId(mqid);
			Log.warn(new EdgeException(mqp, "收到的MQ消息未能成功处理（不能被解码）被抛弃"));
		}
	}

	MQQueueListenerContextImpl getListenerContext() {
		return listenerContext;
	}

	public void afterPropertiesSet() throws Exception {
		Assert.notNull(getQmPool());
	}

	public synchronized void stop() {
		setMonitorStatus(MonitorStatus.STOPPED);
	}

}
